package com.rainsoft.center.isec.frame.record.impl;

import com.rainsoft.center.isec.frame.factory.handler.HandlerFactory;
import com.rainsoft.center.isec.frame.handler.AJavaRDDHandler;
import com.rainsoft.center.isec.frame.record.ARecordDataJob;
import com.rainsoft.center.isec.frame.start.PrepareSpark;
import com.rainsoft.center.isec.stream.library.entity.hbase.BaseLibEntity;
import com.rainsoft.center.isec.stream.library.entity.hbase.impl.VidLoginout;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.function.VoidFunction;
import org.apache.spark.streaming.api.java.JavaDStream;

/**
 * @Package com.rainsoft.center.isec.frame.impl
 * @Description:
 * @Author elwyn
 * @Date 2017/12/10 0:22
 * @Email elonyong@163.com
 */
public class VidRecordImpl extends ARecordDataJob<VidLoginout> {
    public VidRecordImpl(Class<VidLoginout> clazz, PrepareSpark prepareSpark) {
        super(clazz, prepareSpark);
    }

    @Override
    public void foreachDStream() {
        JavaDStream<VidLoginout> javaDStream = getJavaDStream();
        javaDStream.foreachRDD(new VoidFunction<JavaRDD<VidLoginout>>() {
            @Override
            public void call(JavaRDD<VidLoginout> vidLoginoutJavaRDD) throws Exception {
                HandlerFactory handlerFactory = HandlerFactory.getFactory(appConfig.getJobClassName());

                AJavaRDDHandler hbaseHandler1 = handlerFactory.getHbaseHandler(appConfig.gethTableName(), appConfig.getFamily());
                AJavaRDDHandler<BaseLibEntity> hbaseHandler2 = handlerFactory.getUDBHandler(appConfig.gethTableName());

                hbaseHandler1.setNext(hbaseHandler2);
                hbaseHandler1.support(vidLoginoutJavaRDD);
            }
        });
    }
}
